package com.qys.livesMall.shop.scheduledTask;

import com.qys.livesMall.shop.scheduledTask.processor.IDelayTaskProcessor;
import com.qys.livesMall.shop.scheduledTask.taksEnum.TaksQueueEnum;
import com.qys.livesMall.shop.scheduledTask.task.abstrctTaks.AbstractDelayTask;
import jakarta.annotation.Resource;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.redisson.codec.JsonJacksonCodec;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

/**
 * @packageName:
 * @className:DelayTaskDispatcher
 * @Description:
 * @author:guomx
 * @date:2023/7/2915:14
 */
@Component
public class DelayTaskDispatcher implements CommandLineRunner {
    private final static Logger log = LoggerFactory.getLogger(DelayTaskDispatcher.class);
    @Resource
    private RedissonClient redissonClient;
    @Resource
    private ApplicationContext applicationContext;
    private static RBlockingQueue<AbstractDelayTask> consumerQueue ;
    private static RDelayedQueue<AbstractDelayTask> producerQueue;
    // 不同任务类型对应了不同处理器
    private static Map<Integer, IDelayTaskProcessor> taskProcessorByTaskTypeMap;

    // 添加任务
    public static void offerTask(AbstractDelayTask abstractDelayTask) {
        if (!taskProcessorByTaskTypeMap.containsKey(abstractDelayTask.getTaskType())) {
            System.out.println("不支持的任务类型");
        } else {
            log.info("开始任务");
            producerQueue.offer(abstractDelayTask, abstractDelayTask.getTimeout(), abstractDelayTask.getTimeUnit());
        }
    }

    @Override
    public void run(String... args) throws Exception {
        log.info("初始化延迟消费队列");
        taskProcessorByTaskTypeMap = new HashMap<>();
        Map<String, IDelayTaskProcessor> processorMap = applicationContext.getBeansOfType(IDelayTaskProcessor.class);
        for (IDelayTaskProcessor processor : processorMap.values()) {
            taskProcessorByTaskTypeMap.put(processor.supportTask().getTaskType(),processor);
        }
        consumerQueue=redissonClient.getBlockingQueue(TaksQueueEnum.DELAUEDQUEUE.getCode(), new JsonJacksonCodec());
        producerQueue = redissonClient.getDelayedQueue(consumerQueue);
        new Thread(() -> {
            log.info("延迟消费队列启动成功");
            while (true) {
                try {
                    // 阻塞获取任务
                    AbstractDelayTask task = consumerQueue.take();
                    // 获取任务对应的处理器
                    IDelayTaskProcessor iDelayTaskProcessor = taskProcessorByTaskTypeMap.get(task.getTaskType());
                    if (Objects.isNull(iDelayTaskProcessor)) {
                        log.error("不支持的任务类型");
                        continue;
                    }
                    // 执行处理，任务异步处理，以免阻塞任务的获取
                    iDelayTaskProcessor.process(task);
                } catch (Exception e) {
                    log.error(e.getMessage());
                    try {
                        // sleep 500ms，防止极端情况cpu空转
                        Thread.sleep(500);
                    } catch (InterruptedException ex) {
                        log.error(ex.getMessage());
                    }
                }
            }
        }, "delayTaskDispatcherThread").start();
    }
}
